package student.convert

import org.apache.flink.api.common.functions.{RichFilterFunction, RichMapFunction, RichReduceFunction}
import org.apache.flink.streaming.api.scala._
import util.Eastmoney.StockData
import util.RandDomData

object TestRichMapFunction {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env
      .addSource(new RandDomData)
      .map(new RichMapFunction[StockData,(Int,Long)] {
        override def map(in: StockData): (Int, Long) = (in.index,1L)
      })
      .keyBy(_._1)
      .reduce(new RichReduceFunction[(Int, Long)] {
        override def reduce(t: (Int, Long), t1: (Int, Long)): (Int, Long) = (t1._1,t._2+t1._2)
      })
      .filter(new RichFilterFunction[(Int, Long)] {
        override def filter(t: (Int, Long)): Boolean = t._2!=0
      })
      .print()

    env.execute()
  }
}
